package main

import (
	"fmt"
	"github.com/apache/pulsar-client-go/pulsar"
	"go_19/core"
	"go_19/initialize"
	"time"
)

func main() {
	core.EmptyCli()
	option := initialize.ClientOption{
		URL:                     "pulsar://111.111.0.60:6650",
		ConnectionTimeout:       time.Second * time.Duration(3),
		OperationTimeout:        time.Second * time.Duration(30),
		MaxConnectionsPerBroker: 1,
	}
	client := initialize.PulsarClient(option)
	client.CreateReader(pulsar.ReaderOptions{
		Topic:"to_topic",
		StartMessageID:pulsar.EarliestMessageID(),
	})



	consumerMessage := make(chan pulsar.ConsumerMessage, 1)
	consumer, err := client.Subscribe(pulsar.ConsumerOptions{
		Topic:            "to_topic",
		SubscriptionName: "ctz_test",
		Type:             pulsar.Shared,
		Name:             "ctz_test",
		MessageChannel:   consumerMessage,
	})
	if err != nil {
		fmt.Errorf("consumer err", err)
		panic("consumer err")
	}
	for cm := range consumerMessage {
		msg := cm.Message
		fmt.Printf("received[%s],content:'%s',msgId:%v\n", time.Now().Format("2006-01-02 15:04:05"), string(msg.Payload()), msg.ID())
		consumer.Ack(msg)
	}
	defer consumer.Close()
}
